package com.atguigu.bigdata.gmall.realtime.sink;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.gmall.realtime.bean.TableProcess;
import com.atguigu.bigdata.gmall.realtime.util.DimUtil;
import com.atguigu.bigdata.gmall.realtime.util.DruidDSUtil;
import com.atguigu.bigdata.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lzc
 * @Date 2022/10/7 10:16
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    
    private DruidDataSource dataSource;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 1. 获取 连接 池
        dataSource = DruidDSUtil.getDataSource();
        
        redisClient = RedisUtil.getRedisClient();
        
    }
    
    // 流中每来一条数据, 这个方法就执行一次.
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context context) throws Exception {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        
        // 1. 写维度数据到 phoenix 中
        writeToPhoenix(data, tp);
        
        // 2. 更新缓存(粗暴: 删除)
        delCache(data, tp);
    }
    
    private void delCache(JSONObject data, TableProcess tp) {
        String key = DimUtil.getKey(tp.getSinkTable(), data.getString("id"));
        
        // 1. 删除缓存: 当这次是更新的时候才删
        if ("update".equals(data.getString("op_type"))) {
            // 如果要删除的 key 在缓存中不存在会怎么样?
            // 没有任何影响, 只是返回一个 0,表示删除成功了 0 个
            redisClient.del(key);
        }
    }
    
    private void writeToPhoenix(JSONObject data, TableProcess tp) throws SQLException {
        // 获取连接
        DruidPooledConnection conn = dataSource.getConnection();
        // 2. 拼接 sql 语句
        // upsert into user(a,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            // 补充字段名
            .append(tp.getSinkColumns())
            .append(")values(")
            // 补充我?号
            .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
            .append(")");
        
        System.out.println("phoenix 插入语句:" + sql);
        // 3. 从连接获取一个预处理语句
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3.1 给占位符赋值
        // upsert into dim_base_trademark(id,tm_name)values(?,?)
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            String v = data.getString(column);
            ps.setString(i + 1, v);
        }
        // 4. 执行预处理语句
        ps.execute();
        // 5. 手动提交(phoenix 没有自动提交)
        conn.commit();
        // 6. 关闭预处理语句
        ps.close();
        // 归还
        conn.close(); // 如果是从连接池获取的,则是归还, 如果是 DriverManager 获取的,则是关闭.
    }
}
